package com.niit.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;

/**
 * 指定偏移量消费 和 从开始位置消费的 原理是一样
 */

public class CustomConsumerSeek {

    public static void main(String[] args) {

        //1.创建Kafka消费者的连接
        Properties props = new Properties();
        //1.1配置连接机器地址
        props.put("bootstrap.servers","node1:9092");
        //1.2指定消费者组（可以使用消费则组将若干消费者组织到一起），共同消费主题中的数据
        //每一个消费者都需要指定一个消费者组，如果消费者的组名相同，表示这几个消费者是一组的
        props.put("group.id","T3");
        //手动提交 offset
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//从头消费
        //2.创建Kafka的消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);

        //3.订阅要消费的主题
        consumer.subscribe(Arrays.asList("BD1_1"));

        //4.1获得当前主题下所有分区信息
        Set<TopicPartition> assignment = consumer.assignment();
        //4.2初始化消费者分区分配信息 （有了分区的分配信息才能开始消费）
        while (assignment.size() == 0){
            consumer.poll(Duration.ofSeconds(1));
            //将初始化后的分区分配信息重新赋值给 assignment
            assignment = consumer.assignment();
        }
        //4.3遍历所有分区，并指定offset从600开始消费
        for (TopicPartition tp : assignment){
            consumer.seek(tp,74488);
        }

        //5.消费数据
        while (true){
           ConsumerRecords<String,String> records =  consumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String,String> record :records) {
                long offset = record.offset();
                System.out.println("消费的偏移量为："+offset);
            }

        }

    }
}
